package cn.edu.hit.discovery.impl;

import cn.edu.hit.discovery.ServiceDiscovery;
import cn.edu.hit.Constant;
import cn.edu.hit.util.CollectionUtil;
import io.netty.util.internal.ThreadLocalRandom;
import org.I0Itec.zkclient.ZkClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

/**
 * Created by 朱才海 on 2017/12/18.
 */
public class ZookeeperServiceDiscovery implements ServiceDiscovery{

    private static final Logger LOGGER = LoggerFactory.getLogger(ZookeeperServiceDiscovery.class);

    private String zkAddress;

    public ZookeeperServiceDiscovery(String zkAddress){
        this.zkAddress = zkAddress;
    }

    public String discover(String serviceName) {
        //创建 Zookeeper 客户端
        ZkClient zkClient = new ZkClient(zkAddress, Constant.ZK_SESSION_TIMEOUT, Constant.ZK_CONNECTION_TIMEOUT);
        LOGGER.debug("connect to zookeeper successfully");

        try {
            String servicePath = Constant.ZK_REGISTRY_PATH + "/" + serviceName;
            if (!zkClient.exists(servicePath)) {
                throw new RuntimeException("can not find any service node on path:" + servicePath);
            }
            List<String> addressList = zkClient.getChildren(servicePath);
            if (CollectionUtil.isEmpty(addressList)) {
                throw new RuntimeException("can not find any address node on path:" + servicePath);
            }

            //获取 address 节点
            String address;
            int size = addressList.size();
            if (size == 1) {
                //若只有一个地址，则获取该地址
                address = addressList.get(0);
                LOGGER.debug("get the only address node:" + address);
            } else {
                //存在多个地址，则随机获取一个地址
                address = addressList.get(ThreadLocalRandom.current().nextInt(size));
                LOGGER.debug("get random address node:" + address);
            }
            //获取 address 节点的值
            String addressPath = servicePath + "/" + address;
            return zkClient.readData(addressPath);
        }finally {
            zkClient.close();
        }
    }
}
